package com.wpp.register;

import com.wpp.constant.Constant;



import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.util.CollectionUtils;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 *
 */
public class ZkRegisterCenter implements RegisterCenter{


    private CuratorFramework curatorFramework;

    public CuratorFramework getCuratorFramework() {
        return curatorFramework;
    }

    public ZkRegisterCenter() {
        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);
        // 使用fluent编程风格
        this.curatorFramework = CuratorFrameworkFactory.builder()
                .connectString(Constant.ZkConfig.IP_LOCALHOST)
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(exponentialBackoffRetry)
//                .namespace(Constant.ZkConfig.NOTE_ROOT)  // 独立的命名空间 /base
                .build();


        // 启动
        this.curatorFramework.start();
    }

    @Override
    public void register(String serviceName, String serviceAddress) {
        // 注册的服务名 节点路径
        String servicePath = Constant.ZkConfig.REGISTER_ROOT + "/" + serviceName;

        try {
            if (curatorFramework.checkExists().forPath(servicePath) == null) {
                curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(servicePath);
                System.out.printf("创建节点 %s %n ",servicePath);
            }
            // 在服务名节点下创建 服务地址节点（ip + port) 临时节点 服务器 启动 关闭的时候 动态变化
            serviceAddress = servicePath  + "/" +serviceAddress;
            if (curatorFramework.checkExists().forPath(serviceAddress) == null) {
                curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(serviceAddress);
            }
            System.out.printf("服务名: %s下注册了一个服务为：%s %n",serviceName,serviceAddress);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Override
    public Set<ServerInfo> getServerList(String serviceName) {
        Set<ServerInfo> serverInfoSet = new HashSet<>();
        String servicePath = Constant.ZkConfig.REGISTER_ROOT + "/" + serviceName;
        try {
            List<String> serviceList = curatorFramework.getChildren().forPath(servicePath);
            serverInfoSet =  serviceList.stream().map(ZkRegisterCenter::convert).collect(Collectors.toSet());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return serverInfoSet;
    }

    public static ServerInfo convert(String serverStr){
        String[] split = serverStr.split(":");
        ServerInfo serverInfo = new ServerInfo(split[0],Integer.parseInt(split[1]));
        return serverInfo;
    }

}
